package com.jcwy.common.listener;

import cn.hutool.json.JSONUtil;
import com.jcwy.common.IMysqlSyncPlugin;
import com.jcwy.common.dto.MsgDto;
import com.jcwy.common.util.SpringContextUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Optional;

/**
 * @Author: wjs
 * @CreateTime: 2023-03-03 11:46
 */
@Component
@Slf4j
public class CustomSink extends RichSinkFunction<String> {

//    @Lazy
//    @Resource
//    private Map<String, IMysqlSyncPlugin> mysqlSyncPluginMap;

    @Override
    public void open(Configuration parameters) throws Exception {
    }

    @Override
    public void close() throws Exception {
    }

    /**
     * 每次监听到数据变更的时候会走这里
     *
     * @param value
     * @param context
     * @throws Exception
     */
    @Override
    public void invoke(String value, Context context) throws Exception {
        log.info("监听到数据：{}", value);
        Map<String, IMysqlSyncPlugin> mysqlSyncPluginMap = (Map<String, IMysqlSyncPlugin>) SpringContextUtil.getSpringContext().getBean("mysqlSyncPluginMap");

        MsgDto msgDto = JSONUtil.toBean(value, MsgDto.class);
        if (Optional.ofNullable(msgDto.getAfter()).isPresent()) {
            msgDto.setData(JSONUtil.toJsonStr(msgDto.getAfter()));
        } else {
            msgDto.setData(JSONUtil.toJsonStr(msgDto.getBefore()));
        }

        String table = msgDto.getSource().getTable();
        if (!mysqlSyncPluginMap.containsKey(table)) {
            log.error("该数据表：{}没有注册", table);
            return;
        }

        IMysqlSyncPlugin plugin =
                mysqlSyncPluginMap.get(table);
        plugin.process(msgDto);
    }
}
